package com.es.demo.main.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.es.demo.main.IResultHandler;
import com.es.demo.main.cache.EsServer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.IndexNotFoundException;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;


public class ResultHandler_kafka implements IResultHandler<ConsumerRecords<String, String>> {

    @Override
    public void resultHandle(ConsumerRecords<String, String> consumerRecords) {
        for (ConsumerRecord<String, String> record : consumerRecords) {
            /**
             * {
             * 	"data": [{
             * 		"ID": "10",
             * 		"X": "2019-08-12 15:17:54"
             *        }],
             * 	"database": "test",
             * 	"cache": 1565594274000,
             * 	"id": 1,
             * 	"isDdl": false,
             * 	"mysqlType": {
             * 		"ID": "int(11)",
             * 		"X": "timestamp"
             *    },
             * 	"old": null,
             * 	"pkNames": ["ID"],
             * 	"sql": "",
             * 	"sqlType": {
             * 		"ID": 4,
             * 		"X": 93
             *    },
             * 	"table": "xdual",
             * 	"ts": 1565594330790,
             * 	"type": "INSERT"
             * }
             */
            System.err.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            /**
             * 查询es中是否有对应的index
             * 没有则创建index、mapping
             */
            EsServer esServer = new EsServer();
            try {
                esServer.init("my-application-1", Arrays.asList("127.0.0.1:9300", "127.0.0.1:9400"));
            } catch (Exception e) {
                e.printStackTrace();
            }

            /**
             * topic的名字就是dbName_tableName
             * kafka开启了自动创建topic功能，canal的创建topic的规则就是dbName_tableName
             */
            String indexName = record.topic();
            /**
             * 这里获取对应名字的index
             * 1、如果存在，做对应的数据操作
             * 2、如果不存在，创建index、mapping,并做对应的数据操作
             */
            boolean isExists = esServer.checkIndex(indexName);
            System.err.println(isExists);
            if (!isExists) {
                /**创建index*/
                esServer.createIndex(indexName, 0, 0);
                /**创建mapping*/
                Map<String, Map> mapping = new HashMap<>();
                Map<String, Map> childMap = new HashMap<>();
                Map<String, String> paramMap = new HashMap<>();

                paramMap.put("type", "integer");
                childMap.put("id", paramMap);
                paramMap = new HashMap<>();
                paramMap.put("type", "date");
                childMap.put("x", paramMap);

                mapping.put("properties", childMap);

                esServer.createMapping(indexName, null, JSON.toJSONString(mapping));
            }

            /**做对应的数据操作*/


        }
    }
}
